#pragma once
#include <thread>
#include"readerwriterqueue.h"
#include <vector>
using namespace moodycamel;

template<typename T>
class TaskQueue
{
public:
    //forbidden copy & move
	TaskQueue(TaskQueue &&) = delete;
	TaskQueue& operator=(TaskQueue &&) = delete;
    TaskQueue(TaskQueue &) = delete;
    TaskQueue& operator=(const TaskQueue &) = delete;

    //row_num: Number of A scans per file
    //max_buffer: 
    TaskQueue(size_t max_buffer,
        void(*task)(T&&)) :
        queue_(max_buffer), task_(task){};
    ~TaskQueue() {
        stopConsumer();
    };

    //Method
    void startConsumer() {
        if (loop_) return;
        loop_ = true;
        consumer_ = std::thread([this] {
            if (task_ == nullptr) return;
            while (loop_)
            {
                T d;
                if (queue_.try_dequeue(d))
                {
					task_(std::move(d));
                }
                else
                {
					//wait for one mill sec
                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
                }
            }
        });
    } ;
    void stopConsumer(){
        if (!loop_) return;
        loop_ = false;
        if (consumer_.joinable())
            consumer_.join();
    };
    //enqueue input data
    bool post(T&& input) {
        return queue_.enqueue(std::forward<T>(input));
    };

private:
    ReaderWriterQueue<T> queue_;
    std::thread consumer_;
    void(*task_)(T&&) ;
    bool loop_ = false;
};

template<typename T>
class BufferedQueue
{
public:
	BufferedQueue() = delete;
	//forbidden copy move
	BufferedQueue(BufferedQueue &&) = delete;
	BufferedQueue& operator=(BufferedQueue &&) = delete;
	BufferedQueue(BufferedQueue &) = delete;
	BufferedQueue& operator=(const BufferedQueue &) = delete;

	//row_num: Number of A scans per file
	//max_buffer: 
	BufferedQueue(uint32_t row_num, size_t max_buffer,
		void(*task)(std::vector<T>&)) :
		queue_(max_buffer),buffer_(std::vector<T>()),  task_(task), row_num_(row_num) {};
	~BufferedQueue() {
		stopConsumer();
	};

	//Method
	void startConsumer() {
		if (loop_) return;
		loop_ = true;
		consumer_ = std::thread([this] {
			if (task_ == nullptr) return;
			while (loop_)
			{
				T d;
				if (queue_.try_dequeue(d))
				{
					buffer_.emplace_back(std::move(d));
					if(buffer_.size()>=row_num_)
					{
						task_(buffer_);
						buffer_.clear();
						buffer_.reserve(row_num_);
					}
				}
				else
				{
					//wait for one mill sec
					std::this_thread::sleep_for(std::chrono::milliseconds(1));
				}
			}
			if(buffer_.size()>0)
			{
				task_(buffer_);
			}
		});
	};
	void stopConsumer() {
		if (!loop_) return;
		loop_ = false;
		if (consumer_.joinable())
			consumer_.join();
	};
	//enqueue input data
	bool post(T&& input) {
		return queue_.enqueue(std::forward<T>(input));
	};
	//enqueue input data
	bool post(T const&  input) {
		return queue_.enqueue(input);
	};

private:
	ReaderWriterQueue<T> queue_;
	std::thread consumer_;
	std::vector<T> buffer_;
	void(*task_)(std::vector<T>&);
	bool loop_ = false;
	uint32_t row_num_;
};

